package mns

import (
	"reflect"

	. "go-share/lib"

	"github.com/souriki/ali_mns"
	"go-share/lib/crypto"
	"go-share/logging"
)

var (
	client      ali_mns.MNSClient
	environment string
)

// 初始化
func Init(env string, mns interface{}) {
	environment = env
	if client == nil {
		value := reflect.ValueOf(mns)

		accessKeyId := value.FieldByName("AccessKeyId").String()
		accessKeySecret := value.FieldByName("AccessKeySecret").String()

		url := value.FieldByName("Url").String()
		queue := value.FieldByName("Queues")

		client = ali_mns.NewAliMNSClient(url, accessKeyId, accessKeySecret)

		if queue.Kind() == reflect.Struct {
			for i := 0; i < queue.NumField(); i++ {
				v := queue.Field(i).String()
				if v != "" {
					CreateQueue(v)
				}
			}
		}
	}
}

// 创建对列
func CreateQueue(name string) error {
	qm := ali_mns.NewMNSQueueManager(client)
	var queues = [4]string{ENVIRONMENT_MODEL_DEVELOPMENT, ENVIRONMENT_MODEL_PRE, ENVIRONMENT_MODEL_TESTING, ""}
	for _, v := range queues {
		queueName := name
		if v != "" {
			queueName += "-" + v
		}
		err := qm.CreateSimpleQueue(queueName)
		if err != nil &&
			!(ali_mns.ERR_MNS_QUEUE_ALREADY_EXIST.IsEqual(err) ||
				ali_mns.ERR_MNS_QUEUE_ALREADY_EXIST_AND_HAVE_SAME_ATTR.IsEqual(err)) {
			logging.Panic("Create queue | %v", err)
		} else {
			logging.Debug("Create queue %v Success", queueName)
		}
	}
	logging.Info("All queue creation is complete")

	return nil
}

// 创建对列
func CreateSimpleQueue(name string) error {
	qm := ali_mns.NewMNSQueueManager(client)
	err := qm.CreateSimpleQueue(name)
	if err != nil && !ali_mns.ERR_MNS_QUEUE_ALREADY_EXIST_AND_HAVE_SAME_ATTR.IsEqual(err) {
		return err
	}
	return nil
}

// 发布消息
func Publish(queueName string, body []byte, delay int64, priority int64) error {
	queueName = getQueueName(queueName)
	msg := ali_mns.MessageSendRequest{
		MessageBody:  string(crypto.EncodeBase64(body)),
		DelaySeconds: delay,
		Priority:     priority}

	q := ali_mns.NewMNSQueue(queueName, client)
	_, err := q.SendMessage(msg)

	logging.Debug("Send message to the %v queue", queueName)
	return err
}

// 订阅消息
func Subscribe(queueName string, callback func(msg []byte, err error)) {
	queueName = getQueueName(queueName)
	queue := ali_mns.NewMNSQueue(queueName, client)
	respChan := make(chan ali_mns.MessageReceiveResponse)
	errChan := make(chan error)

	logging.Debug("Subscribe to the %v queue", queueName)

	go func() {
		for {
			select {
			case resp := <-respChan:
				b, _ := crypto.DecodeBase64([]byte(resp.MessageBody))
				if err := queue.DeleteMessage(resp.ReceiptHandle); err != nil {
					callback(nil, err)
				} else {
					callback(b, nil)
				}

			case err := <-errChan:
				if ali_mns.ERR_MNS_QUEUE_NOT_EXIST.IsEqual(err) {
					close(errChan)
					close(respChan)
					callback(nil, err)
					return
				}

				if ali_mns.ERR_MNS_MESSAGE_NOT_EXIST.IsEqual(err) {
					// callback(nil, err)
					// time.Sleep(time.Second * 10)
				}
			}
		}
	}()

	for {
		if respChan == nil {
			return
		}
		queue.ReceiveMessage(respChan, errChan, 30)
	}
}

// --------------------------------------------------------------------------------

func getQueueName(name string) string {
	if environment == ENVIRONMENT_MODEL_PRODUCTION || environment == "" {
		return name
	}
	name += "-" + environment
	return name
}
